package com.yongche;

import java.util.Iterator;

import org.apache.spark.SparkConf;
import org.apache.spark.api.java.JavaPairRDD;
import org.apache.spark.api.java.JavaRDD;
import org.apache.spark.api.java.JavaSparkContext;
import org.apache.spark.api.java.function.Function;
import org.apache.spark.api.java.function.Function2;
import org.apache.spark.api.java.function.PairFunction;
import org.apache.spark.sql.DataFrame;
import org.apache.spark.sql.SQLContext;
import org.apache.spark.storage.StorageLevel;

import scala.Tuple2;

import com.yongche.bigdata.model.metric.OriginAmountMetric;
import com.yongche.bigdata.model.vote.OriginAmountVote;

public class OriginAmount_Offline {

	public static void main(String[] args) {
		
	SparkConf conf = new SparkConf()
	.set("spark.streaming.unpersist", "true") //Spark来计算哪些RDD需要持久化，这样有利于提高GC的表现。
	.set("spark.default.parallelism", "8")	//reduceByKey 执行时启动的线程数，默认是8个
	.set("spark.storage.memoryFraction", "0.5")
	.set("spark.shuffle.consolidateFiles", "true");
	
    JavaSparkContext context = new JavaSparkContext(conf);
    final SQLContext sqlcontext = new SQLContext(context);
   
    JavaRDD<String> messages = context.textFile("/user/zhaoxiang/data/service_order/*.data");

    JavaPairRDD<String, OriginAmountMetric> orderLines = messages.mapToPair(new PairFunction<String, String, OriginAmountMetric>() {
		private static final long serialVersionUID = 1L;
		@Override
		public Tuple2<String, OriginAmountMetric> call(String t) throws Exception {
			String line = t;
            String[] items = line.split("\001");
            String id = items[0];
            String city = items[18];
            int status = Integer.parseInt(items[6]);
            long flag =Long.parseLong(items[10]);
            long f = flag & 0x400000;
            double origin_amount = Double.parseDouble(items[73]);
            double origin_sharing_amount = Double.parseDouble(items[74]);
            int pay_amount = (int) Double.parseDouble(items[71]);
            String op_type = items[86];
            if(status==7 && f==0x400000 && pay_amount==0 && !"DELETE".equals(op_type)){
            	return new Tuple2<String, OriginAmountMetric>(id, new OriginAmountMetric(id, status, origin_sharing_amount,origin_amount, flag, city));
            }
            return new Tuple2<String, OriginAmountMetric>(id, null);
        }
	}).filter(new Function<Tuple2<String, OriginAmountMetric>, Boolean>() {
    	private static final long serialVersionUID = 1L;
		@Override
		public Boolean call(Tuple2<String, OriginAmountMetric> t) throws Exception {
            return t._2() != null;
		}
    });

    JavaPairRDD<String, OriginAmountMetric> orders = orderLines.reduceByKey(new Function2<OriginAmountMetric, OriginAmountMetric, OriginAmountMetric>() {
    	private static final long serialVersionUID = 1L;
		@Override
		public OriginAmountMetric call(OriginAmountMetric v1,
				OriginAmountMetric v2) throws Exception {
			return v2;
		}
	});
    
    JavaPairRDD<String, OriginAmountMetric> orderPairs = orders.mapToPair(new PairFunction<Tuple2<String,OriginAmountMetric>, String, OriginAmountMetric>() {
    	private static final long serialVersionUID = 1L;
		@Override
		public Tuple2<String, OriginAmountMetric> call(
				Tuple2<String, OriginAmountMetric> t) throws Exception {
			return new Tuple2<String, OriginAmountMetric>(t._2.getCity(), t._2);
		}
    });

    JavaRDD<OriginAmountVote> cs = orderPairs.groupByKey().map(new Function<Tuple2<String,Iterable<OriginAmountMetric>>, OriginAmountVote>() {
    	private static final long serialVersionUID = 1L;
        @Override
        public OriginAmountVote call(Tuple2<String, Iterable<OriginAmountMetric>> t) throws Exception {
            String city = t._1();
            Iterator<OriginAmountMetric> it = t._2().iterator();
            double amount = 0.0;
            while(it.hasNext()) {
                OriginAmountMetric m = it.next();
                amount += m.getOriginSharingAmount() - m.getOriginAmount();;

            }
            amount = (double)Math.round(amount*1000)/1000;
            return new OriginAmountVote(city, amount);
        }
    }).persist(StorageLevel.MEMORY_AND_DISK_SER());

    DataFrame schema = sqlcontext.createDataFrame(cs, OriginAmountVote.class);
    schema.registerTempTable("t_origin_amount_vote");
    DataFrame df = sqlcontext.sql("select 'quanguo', sum(amount) from t_origin_amount_vote");
    df.show();
    for (OriginAmountVote vo : cs.take(500)){
    	System.out.println(vo);
    };
    cs.unpersist();
    context.close();
//    JavaRDD<OriginAmountVote> rs =  df.javaRDD().map(new Function<Row, OriginAmountVote>() {
//    	private static final long serialVersionUID = 1L;
//        @Override
//        public OriginAmountVote call(Row row) throws Exception {
//            String city = row.getString(0);
//            double amount = row.getDouble(1);
//            amount = (double)Math.round(amount*1000)/1000;
//            return new OriginAmountVote(city, amount);
//        }
//    });
//    
//    JavaRDD<OriginAmountVote> qs = 
//           
//            return rs;
//        }
//    });
}


}
